package com.moli.iov.gateway.server.listener;

import com.moli.iov.consumer.KafkaConsumerServer;
import com.moli.iov.gateway.server.task.DownStreamDataTask;
import com.moli.iov.gateway.server.VinThreadPoolExecutor;
import com.moli.iov.protocol.codec.DataWrapper;
import com.moli.iov.protocol.codec.DownStreamData;
import com.moli.iov.util.DateUtil;
import com.moli.iov.util.StringUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.support.Acknowledgment;

import javax.annotation.Resource;

/**
 * <p>
 * 下发命令处理
 * </p>
 *
 * @Auther: zql
 * @Date: 2018/12/3 14:36
 */
public class ConsumerMessageListener extends KafkaConsumerServer {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerMessageListener.class);

    @Resource(name = "downThreadPool")
    private VinThreadPoolExecutor downThreadPool;
    /**
     * 下发命令最大时间 10 分钟
     * 超过10分钟，则不下发指令
     */
    @Value("${send_max_time}")
    private long sendMaxTime;

    @Override
    public void manualHandle(ConsumerRecord<String, Object> record, Acknowledgment acknowledgment) {
        String key = record.key();
        Object value = record.value();
        try {

            DataWrapper dataWrapper = (DataWrapper) value;
            if (dataWrapper == null) {
                logger.error("[下行报文] [下发命令处理失败] value为null  key：{}", key);
                return;
            }
            if (dataWrapper.getSendTime() == null) {
                logger.error("[下行报文] [下发命令处理失败] sendTime为null  key：{}, value:{}", key, dataWrapper);
                return;
            }

            if (System.currentTimeMillis() - dataWrapper.getSendTime().getTime() > sendMaxTime) {
                logger.error("[下行报文] [下发命令处理失败] [超过下发命令最大时间] 上层下发时间：{}， 最大小时间：{}", DateUtil.format("yyyy-MM-dd HH:mm:ss:SSS", dataWrapper.getSendTime()), sendMaxTime);
                return;
            }
            logger.info("[下行报文] [下发命令处理] key：{}, data:{},  value: {}", key, dataWrapper, StringUtil.bytesToHexString(dataWrapper.getAllData()));
            DownStreamData dsd = new DownStreamData(key, dataWrapper.getAllData(), dataWrapper.getSerialNo());
            downThreadPool.execute(key, new DownStreamDataTask(dsd));
        } catch (Exception e) {
            logger.error(String.format("[下行报文] [下发命令处理异常]  vin: %s, value: %s", key, value), e);
        }
        acknowledgment.acknowledge();
    }
}
